package com.hkbigdata.watermark;

import com.hkbigdata.entry.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author liuanbo
 * @creat 2023-04-24-17:46
 * @see 2194550857@qq.com
 */
public class Flink03_Process_SideOutPut {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取端口数据并转换为JavaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("hadoop102", 9999)
                .map(data -> {
                    String[] split = data.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                });

        SingleOutputStreamOperator<WaterSensor> process = waterSensorDS.process(new ProcessFunction<WaterSensor, WaterSensor>() {
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                //1.取出水位线
                Integer vs = value.getVs();

                if (vs >= 30) {
                    out.collect(value);
                } else {
                    ctx.output(new OutputTag<Tuple2<String,Integer>>("Side") {
                    }, new Tuple2<>(value.getId(), vs));
                }
            }
        });

        process.print("主流");
        process.getSideOutput(new OutputTag<Tuple2<String,Integer>>("Side"){}).print("side");

        env.execute();
    }
}
